import os
import time
import logging
from datetime import datetime, timedelta
from dotenv import load_dotenv
import baostock as bs
import pymysql
from apscheduler.schedulers.background import BackgroundScheduler

# 加载环境变量
load_dotenv()

# 日志配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(module)s:%(lineno)d - %(message)s',
    handlers=[
        logging.FileHandler('stock_k_sync.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# 数据库配置
DB_CONFIG = {
    'host': os.getenv('DB_HOST', 'localhost'),
    'user': os.getenv('DB_USER', 'root'),
    'password': os.getenv('DB_PASSWORD', '123456'),
    'database': os.getenv('DB_NAME', 'stock_data'),
    'port': int(os.getenv('DB_PORT', 3306)),
    'charset': 'utf8mb4'
}

# 路径配置
PROGRESS_FILE_DAY = 'synced_daily_stocks.txt'
PROGRESS_FILE_WEEK = 'synced_weekly.txt'
PROGRESS_FILE_MONTH = 'synced_monthly.txt'
STOCK_CODE_CACHE = 'stock_codes_cache.txt'
CACHE_EXPIRE_HOURS = 24

# 批量处理参数
BATCH_SIZE = 100
API_INTERVAL_DAY = 0.5
STOCK_RETRY_COUNT = 5
MAX_TRADING_DAY_LOOKBACK = 10
CUTOFF_HOUR = 17
CUTOFF_MINUTE = 30

# 全局登录状态标记
baostock_logged_in = False


def ensure_baostock_login():
    """确保Baostock处于登录状态，失败则重试"""
    global baostock_logged_in
    if baostock_logged_in:
        return True

    for i in range(STOCK_RETRY_COUNT):
        try:
            bs.logout()
            time.sleep(0.5)
            login_status = bs.login()
            if login_status.error_code == '0':
                baostock_logged_in = True
                logger.info("Baostock登录成功")
                return True
            logger.warning(f"Baostock登录失败（第{i + 1}次）：{login_status.error_msg}")
            time.sleep(2 * (i + 1))
        except Exception as e:
            logger.error(f"登录过程异常（第{i + 1}次）：{str(e)}")
            time.sleep(2 * (i + 1))

    logger.error("Baostock登录失败，无法继续操作")
    return False


def baostock_safe_logout():
    """安全登出"""
    global baostock_logged_in
    if baostock_logged_in:
        try:
            bs.logout()
            logger.info("Baostock安全登出")
        except Exception as e:
            logger.warning(f"登出时发生异常：{str(e)}")
        finally:
            baostock_logged_in = False


# 数据转换函数（仅日K需要）
def convert_volume(volume_str, code=None):
    if not volume_str or volume_str.strip() in ['', '0']:
        return 0.0
    try:
        volume = int(volume_str.strip())
        return round(volume / 1000000, 4)  # 股 → 万手
    except (TypeError, ValueError) as e:
        logger.warning(f"成交量转换失败（{volume_str}）：{str(e)}")
        return None


def convert_amount(amount_str, code=None):
    if not amount_str or amount_str.strip() in ['', '0']:
        return 0.0
    try:
        amount = float(amount_str.strip())
        return round(amount / 100000000, 4)  # 元 → 亿元
    except (TypeError, ValueError) as e:
        logger.warning(f"成交额转换失败（{amount_str}）：{str(e)}")
        return None


# 股票代码管理（仅用于日K同步，周/月K无需循环股票）
def is_cache_expired(cache_file):
    if not os.path.exists(cache_file):
        return True
    cache_mtime = os.path.getmtime(cache_file)
    return (time.time() - cache_mtime) > (CACHE_EXPIRE_HOURS * 3600)


def get_latest_trading_day(force_check_today=False, use_cutoff=True):
    """获取最近的交易日"""
    try:
        now = datetime.now()
        is_after_cutoff = (now.hour > CUTOFF_HOUR) or (now.hour == CUTOFF_HOUR and now.minute >= CUTOFF_MINUTE)

        base_date = now if (force_check_today or is_after_cutoff) else now - timedelta(days=1)

        logger.info(
            f"当前时间{now.strftime('%H:%M')}（{'已过' if is_after_cutoff else '未过'}17:30），基准检查日期：{base_date.strftime('%Y-%m-%d')}")

        for i in range(MAX_TRADING_DAY_LOOKBACK):
            current_date = (base_date - timedelta(days=i)).strftime('%Y-%m-%d')
            current_weekday = datetime.strptime(current_date, '%Y-%m-%d').weekday()
            if current_weekday >= 5:
                logger.debug(f"日期{current_date}为周末，跳过")
                continue
            if is_trading_day(current_date):
                logger.info(f"找到最近的有效交易日：{current_date}")
                return current_date

        logger.error(f"向前追溯{MAX_TRADING_DAY_LOOKBACK}天未找到交易日")
        return None
    except Exception as e:
        logger.error(f"获取最近交易日失败：{str(e)}", exc_info=True)
        return None


def fetch_stock_codes_from_baostock(trading_day):
    """获取股票列表（仅日K同步用）"""
    if not ensure_baostock_login():
        return []

    try:
        rs = bs.query_all_stock(trading_day)
        if rs.error_code != '0':
            logger.error(f"获取股票列表失败（交易日{trading_day}）：{rs.error_msg}")
            if '网络' in rs.error_msg or '连接' in rs.error_msg and ensure_baostock_login():
                return fetch_stock_codes_from_baostock(trading_day)
            return []

        stock_codes = []
        while (rs.error_code == '0') and rs.next():
            code = rs.get_row_data()[0]
            if code.startswith(('sh.60', 'sh.688', 'sz.00', 'sz.30', 'bj.8')):
                stock_codes.append(code)

        logger.info(f"从Baostock获取到{len(stock_codes)}只A股股票（交易日：{trading_day}）")
        return stock_codes
    except Exception as e:
        logger.error(f"获取股票列表异常：{str(e)}", exc_info=True)
        if '网络' in str(e) or 'socket' in str(e).lower() and ensure_baostock_login():
            return fetch_stock_codes_from_baostock(trading_day)
        return []


def update_stock_code_cache():
    logger.info("=== 开始更新股票列表缓存 ===")
    try:
        latest_trading_day = get_latest_trading_day(force_check_today=False, use_cutoff=True)
        if not latest_trading_day:
            logger.error("无法获取有效交易日，股票列表缓存更新失败")
            return False

        stock_codes = []
        for retry in range(STOCK_RETRY_COUNT):
            stock_codes = fetch_stock_codes_from_baostock(latest_trading_day)
            if stock_codes:
                break
            logger.warning(f"第{retry + 1}次获取股票列表失败，重试...")
            time.sleep(3 * (retry + 1))

        if not stock_codes:
            logger.error("获取股票列表为空，缓存更新失败")
            return False

        with open(STOCK_CODE_CACHE, 'w', encoding='utf-8') as f:
            f.write(f"# 缓存时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
            f.write(f"# 对应交易日：{latest_trading_day}\n")
            f.write('\n'.join(stock_codes))

        logger.info(f"股票列表缓存更新成功（缓存文件：{STOCK_CODE_CACHE}）")
        return True
    except Exception as e:
        logger.error(f"更新股票缓存失败：{str(e)}", exc_info=True)
        return False


def get_stock_codes():
    try:
        if not is_cache_expired(STOCK_CODE_CACHE):
            logger.info("使用有效期内的股票列表缓存")
            with open(STOCK_CODE_CACHE, 'r', encoding='utf-8') as f:
                return [line.strip() for line in f if line.strip() and not line.startswith('#')]

        if not update_stock_code_cache():
            logger.warning("缓存更新失败，尝试从数据库提取历史股票列表")
            return get_stocks_from_local_db()

        with open(STOCK_CODE_CACHE, 'r', encoding='utf-8') as f:
            return [line.strip() for line in f if line.strip() and not line.startswith('#')]
    except Exception as e:
        logger.error(f"获取股票代码列表失败：{str(e)}", exc_info=True)
        return []


# 交易日工具
def is_trading_day(date_str):
    """检查是否为交易日"""
    if not ensure_baostock_login():
        return False

    try:
        rs = bs.query_trade_dates(start_date=date_str, end_date=date_str)
        if rs.error_code != '0':
            logger.warning(f"查询日期{date_str}失败：{rs.error_msg}")
            if '网络' in rs.error_msg or '连接' in rs.error_msg and ensure_baostock_login():
                return is_trading_day(date_str)
            return False

        if rs.next():
            _, is_trading = rs.get_row_data()
            return is_trading == '1'
        return False
    except Exception as e:
        logger.error(f"检查交易日异常：{str(e)}", exc_info=True)
        if '网络' in str(e) or 'socket' in str(e).lower() and ensure_baostock_login():
            return is_trading_day(date_str)
        return False


# 数据库初始化
def init_database():
    try:
        conn = pymysql.connect(
            host=DB_CONFIG['host'],
            user=DB_CONFIG['user'],
            password=DB_CONFIG['password'],
            port=DB_CONFIG['port'],
            charset=DB_CONFIG['charset']
        )
        cursor = conn.cursor()
        cursor.execute(f"CREATE DATABASE IF NOT EXISTS {DB_CONFIG['database']}")
        conn.select_db(DB_CONFIG['database'])

        # 日K表（优化索引）
        create_day_sql = """
                         CREATE TABLE IF NOT EXISTS stock_day_k_history (
                             id VARCHAR(50) NOT NULL COMMENT '唯一标识（代码_日期）',
                             code VARCHAR(20) NOT NULL COMMENT '股票代码（纯数字）',
                             date DATE NOT NULL COMMENT '日K日期',
                             open DECIMAL(10, 2) NULL COMMENT '开盘价',
                             high DECIMAL(10, 2) NULL COMMENT '最高价',
                             low DECIMAL(10, 2) NULL COMMENT '最低价',
                             close DECIMAL(10, 2) NULL COMMENT '收盘价',
                             last_day_close DECIMAL(10, 2) NULL COMMENT '上一日收盘价',
                             volume DECIMAL(12, 2) NULL COMMENT '成交量（万手）',
                             amount DECIMAL(12, 2) NULL COMMENT '成交额（亿元）',
                             adjustflag TINYINT NULL COMMENT '复权类型（1前复权/2后复权）',
                             turn DECIMAL(10, 2) NULL COMMENT '换手率（%）',
                             pctChg DECIMAL(10, 2) NULL COMMENT '涨跌幅（%）',
                             insert_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                             PRIMARY KEY (id),
                             INDEX idx_code_date (code, date),  -- 优化按code+date的精确查询
                             INDEX idx_date_code (date, code)   -- 优化按date范围+code分组的聚合查询
                         ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
                         """

        # 周K表（优化索引）
        create_week_sql = """
                          CREATE TABLE IF NOT EXISTS stock_week_k_history (
                              id VARCHAR(50) NOT NULL COMMENT '唯一标识（代码_周期末日期）',
                              code VARCHAR(20) NOT NULL COMMENT '股票代码',
                              date DATE NOT NULL COMMENT '周K周期末日期（周五）',
                              open DECIMAL(10, 2) NULL COMMENT '周开盘价',
                              high DECIMAL(10, 2) NULL COMMENT '周最高价',
                              low DECIMAL(10, 2) NULL COMMENT '周最低价',
                              close DECIMAL(10, 2) NULL COMMENT '周收盘价',
                              last_week_close DECIMAL(10, 2) NULL COMMENT '上周收盘价',
                              volume DECIMAL(12, 2) NULL COMMENT '周成交量（万手）',
                              amount DECIMAL(12, 2) NULL COMMENT '周成交额（亿元）',
                              adjustflag TINYINT NULL COMMENT '复权类型',
                              turn DECIMAL(10, 2) NULL COMMENT '周换手率（%）',
                              pctChg DECIMAL(10, 2) NULL COMMENT '周涨跌幅（%）',
                              insert_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                              PRIMARY KEY (id),
                              INDEX idx_date_code (date, code)  -- 优化按date删除/查询
                          ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
                          """

        # 月K表（优化索引）
        create_month_sql = """
                           CREATE TABLE IF NOT EXISTS stock_month_k_history (
                               id VARCHAR(50) NOT NULL COMMENT '唯一标识（代码_周期末日期）',
                               code VARCHAR(20) NOT NULL COMMENT '股票代码',
                               date DATE NOT NULL COMMENT '月K周期末日期（月末）',
                               open DECIMAL(10, 2) NULL COMMENT '月开盘价',
                               high DECIMAL(10, 2) NULL COMMENT '月最高价',
                               low DECIMAL(10, 2) NULL COMMENT '月最低价',
                               close DECIMAL(10, 2) NULL COMMENT '月收盘价',
                               last_month_close DECIMAL(10, 2) NULL COMMENT '上月收盘价',
                               volume DECIMAL(12, 2) NULL COMMENT '月成交量（万手）',
                               amount DECIMAL(12, 2) NULL COMMENT '月成交额（亿元）',
                               adjustflag TINYINT NULL COMMENT '复权类型',
                               turn DECIMAL(10, 2) NULL COMMENT '月换手率（%）',
                               pctChg DECIMAL(10, 2) NULL COMMENT '月涨跌幅（%）',
                               insert_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                               PRIMARY KEY (id),
                               INDEX idx_date_code (date, code)  -- 优化按date删除/查询
                           ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
                           """

        cursor.execute(create_day_sql)
        cursor.execute(create_week_sql)
        cursor.execute(create_month_sql)
        conn.commit()
        logger.info("数据库表结构（含优化索引）初始化成功")
    except Exception as e:
        logger.error(f"数据库初始化失败：{str(e)}", exc_info=True)
        raise
    finally:
        if 'conn' in locals():
            conn.close()


def get_stocks_from_local_db():
    """从数据库获取历史股票列表（仅日K同步用）"""
    try:
        conn = pymysql.connect(**DB_CONFIG)
        cursor = conn.cursor()
        cursor.execute("""
                       SELECT DISTINCT code FROM stock_day_k_history
                       UNION
                       SELECT DISTINCT code FROM stock_week_k_history
                       UNION
                       SELECT DISTINCT code FROM stock_month_k_history
                       """)
        raw_codes = [row[0] for row in cursor.fetchall()]
        if not raw_codes:
            logger.warning("数据库中无历史股票数据")
            return []

        stock_codes = []
        for code in raw_codes:
            if code.startswith('6'):
                stock_codes.append(f'sh.{code}')
            elif code.startswith(('0', '3')):
                stock_codes.append(f'sz.{code}')
            elif code.startswith('8'):
                stock_codes.append(f'bj.{code}')

        logger.info(f"从数据库提取到{len(stock_codes)}只股票")
        return stock_codes
    except Exception as e:
        logger.error(f"数据库提取股票失败：{str(e)}", exc_info=True)
        return []
    finally:
        if 'conn' in locals():
            conn.close()


# 数据格式工具
def clean_stock_code(raw_code):
    return raw_code.split('.')[1] if '.' in raw_code else raw_code


def generate_id(code, date):
    return f"{code}_{date.replace('-', '')}"


# 周期时间工具（计算周/月范围）
def get_week_params():
    """获取周参数：周一至周五实际交易日范围"""
    try:
        today = datetime.now()
        weekday = today.weekday()  # 0=周一, 4=周五
        week_start = (today - timedelta(days=weekday)).strftime('%Y-%m-%d')
        week_end = (today + timedelta(days=4 - weekday)).strftime('%Y-%m-%d')
        last_week_end = (datetime.strptime(week_end, '%Y-%m-%d') - timedelta(days=7)).strftime('%Y-%m-%d')

        # 修正为实际交易日（跳过周末和节假日）
        week_start = week_start if is_trading_day(week_start) else get_next_trading_day(week_start)
        week_end = week_end if is_trading_day(week_end) else get_prev_trading_day(week_end)
        last_week_end = last_week_end if is_trading_day(last_week_end) else get_prev_trading_day(last_week_end)

        return {
            'current': (week_start, week_end),  # 本周实际交易日范围
            'last_week_end': last_week_end,     # 上周最后一个交易日（用于取上周收盘价）
            'is_complete': weekday >= 4,        # 周五及之后视为本周结束
            'week_end': week_end                # 周K周期末日期
        }
    except Exception as e:
        logger.error(f"计算周参数失败: {str(e)}", exc_info=True)
        return {
            'current': ('2025-09-01', '2025-09-05'),
            'last_week_end': '2025-08-29',
            'is_complete': True,
            'week_end': '2025-09-05'
        }


def get_month_params():
    """获取月参数：月初至月末实际交易日范围"""
    try:
        today = datetime.now()
        month_start = today.replace(day=1).strftime('%Y-%m-%d')
        # 计算月末（下个月1号减1天）
        next_month = today.replace(day=28) + timedelta(days=4)
        month_end = (next_month.replace(day=1) - timedelta(days=1)).strftime('%Y-%m-%d')
        # 上月末
        last_month_end = (datetime.strptime(month_start, '%Y-%m-%d') - timedelta(days=1)).strftime('%Y-%m-%d')

        # 修正为实际交易日
        month_start = month_start if is_trading_day(month_start) else get_next_trading_day(month_start)
        month_end = month_end if is_trading_day(month_end) else get_prev_trading_day(month_end)
        last_month_end = last_month_end if is_trading_day(last_month_end) else get_prev_trading_day(last_month_end)

        return {
            'current': (month_start, month_end),  # 本月实际交易日范围
            'last_month_end': last_month_end,     # 上月最后一个交易日
            'is_complete': today >= datetime.strptime(month_end, '%Y-%m-%d'),  # 本月是否结束
            'month_end': month_end                # 月K周期末日期
        }
    except Exception as e:
        logger.error(f"计算月参数失败: {str(e)}", exc_info=True)
        return {
            'current': ('2025-09-01', '2025-09-30'),
            'last_month_end': '2025-08-31',
            'is_complete': False,
            'month_end': '2025-09-30'
        }


def get_next_trading_day(date_str):
    """获取指定日期后的第一个交易日"""
    date_obj = datetime.strptime(date_str, '%Y-%m-%d')
    for i in range(1, 10):
        next_date = (date_obj + timedelta(days=i)).strftime('%Y-%m-%d')
        if is_trading_day(next_date):
            return next_date
    return date_str


def get_prev_trading_day(date_str):
    """获取指定日期前的第一个交易日"""
    date_obj = datetime.strptime(date_str, '%Y-%m-%d')
    for i in range(1, 10):
        prev_date = (date_obj - timedelta(days=i)).strftime('%Y-%m-%d')
        if is_trading_day(prev_date):
            return prev_date
    return date_str


# 日K同步逻辑
def get_last_synced_date():
    if os.path.exists(PROGRESS_FILE_DAY):
        with open(PROGRESS_FILE_DAY, 'r', encoding='utf-8') as f:
            lines = [line.strip() for line in f if line.strip() and line.startswith('LAST_SYNC_DATE:')]
            if lines:
                return lines[0].replace('LAST_SYNC_DATE:', '').strip()
    return None


def update_synced_date(date_str):
    lines = []
    if os.path.exists(PROGRESS_FILE_DAY):
        with open(PROGRESS_FILE_DAY, 'r', encoding='utf-8') as f:
            lines = [line for line in f if not line.startswith('LAST_SYNC_DATE:')]
    with open(PROGRESS_FILE_DAY, 'w', encoding='utf-8') as f:
        f.write(f"LAST_SYNC_DATE:{date_str}\n")
        f.writelines(lines)


def fetch_stock_day_k_today(raw_code, trade_date):
    """获取单只股票当天的日K数据"""
    global baostock_logged_in
    logger.debug(f"获取{raw_code}当天({trade_date})日K数据")

    if not ensure_baostock_login():
        logger.warning(f"Baostock未登录，跳过{raw_code}日K获取")
        return []

    try:
        DAY_K_FIELDS = ['code', 'date', 'open', 'high', 'low', 'close', 'volume', 'amount', 'adjustflag', 'turn', 'pctChg']
        rs = bs.query_history_k_data_plus(
            raw_code,
            ",".join(DAY_K_FIELDS),
            start_date=trade_date,
            end_date=trade_date,
            frequency='d',
            adjustflag='2'
        )

        if rs.error_code != '0':
            logger.warning(f"{raw_code}日K获取失败：{rs.error_msg}")
            if '网络' in rs.error_msg or 'socket' in rs.error_msg.lower() and ensure_baostock_login():
                return fetch_stock_day_k_today(raw_code, trade_date)
            return []

        day_k_list = []
        code = clean_stock_code(raw_code)
        while rs.next():
            row = rs.get_row_data()
            if row[1] != trade_date:
                logger.warning(f"{raw_code}返回日期不匹配：{row[1]} != {trade_date}，数据丢弃")
                continue

            day_k = {
                'id': generate_id(code, row[1]),
                'code': code,
                'date': row[1],
                'open': round(float(row[2]), 2) if row[2] else None,
                'high': round(float(row[3]), 2) if row[3] else None,
                'low': round(float(row[4]), 2) if row[4] else None,
                'close': round(float(row[5]), 2) if row[5] else None,
                'volume': convert_volume(row[6], code=code),
                'amount': convert_amount(row[7], code=code),
                'adjustflag': int(row[8]) if row[8] else None,
                'turn': round(float(row[9]), 2) if row[9] else None,
                'pctChg': round(float(row[10]), 2) if row[10] else None
            }
            day_k_list.append(day_k)

        return day_k_list
    except Exception as e:
        logger.error(f"{raw_code}日K处理异常：{str(e)}", exc_info=True)
        if '10038' in str(e) or 'socket' in str(e).lower():
            baostock_logged_in = False
            if ensure_baostock_login():
                return fetch_stock_day_k_today(raw_code, trade_date)
        return []


def get_previous_trading_day(trade_date):
    try:
        date_obj = datetime.strptime(trade_date, '%Y-%m-%d')
        for i in range(1, MAX_TRADING_DAY_LOOKBACK + 1):
            prev_date = (date_obj - timedelta(days=i)).strftime('%Y-%m-%d')
            if is_trading_day(prev_date):
                return prev_date
        return None
    except Exception as e:
        logger.error(f"获取前一交易日失败：{str(e)}", exc_info=True)
        return None


def get_previous_close(code, prev_trade_date):
    try:
        conn = pymysql.connect(**DB_CONFIG)
        cursor = conn.cursor()
        cursor.execute("""
            SELECT close FROM stock_day_k_history
            WHERE code = %s AND date = %s
        """, (code, prev_trade_date))

        result = cursor.fetchone()
        return result[0] if result else None
    except Exception as e:
        logger.error(f"查询{code}前收盘价失败：{str(e)}", exc_info=True)
        return None
    finally:
        if 'conn' in locals():
            conn.close()


def sync_day_k():
    try:
        logger.info("===== 开始日K同步任务 =====")
        start_time = time.time()
        global baostock_logged_in
        baostock_logged_in = False

        now = datetime.now()
        today = now.strftime('%Y-%m-%d')
        is_after_cutoff = (now.hour > CUTOFF_HOUR) or (now.hour == CUTOFF_HOUR and now.minute >= CUTOFF_MINUTE)

        if not is_after_cutoff:
            logger.info(f"当前时间{now.strftime('%H:%M')}，未过{CUTOFF_HOUR}:{CUTOFF_MINUTE}，暂不同步日K数据")
            return

        latest_trading_day = get_latest_trading_day(force_check_today=True, use_cutoff=False)
        if not latest_trading_day or latest_trading_day != today:
            logger.info(f"今天({today})不是交易日，无需同步日K数据")
            return

        last_synced = get_last_synced_date()
        if last_synced == today:
            logger.info(f"今天({today})的日K数据已经同步过，无需重复同步")
            return

        stock_codes = get_stock_codes()
        if not stock_codes:
            logger.error("无股票列表可同步，日K任务终止")
            return

        if not ensure_baostock_login():
            logger.error("无法登录Baostock，日K任务终止")
            return

        prev_trade_date = get_previous_trading_day(today)
        if not prev_trade_date:
            logger.warning("无法获取前一个交易日，无法计算上一日收盘价")

        batch_data = []
        total_processed = 0
        total_inserted = 0
        network_error_count = 0

        for raw_code in stock_codes:
            if network_error_count >= 5:
                logger.info("检测到连续网络错误，暂停10秒...")
                time.sleep(10)
                network_error_count = 0

            try:
                day_k_list = fetch_stock_day_k_today(raw_code, today)

                if not day_k_list:
                    last_error = logger.handlers[0].buffer[-1] if logger.handlers[0].buffer else ""
                    if '网络' in str(last_error) or 'socket' in str(last_error).lower():
                        network_error_count += 1
                    else:
                        network_error_count = 0
                    continue

                for day_k in day_k_list:
                    if prev_trade_date:
                        day_k['last_day_close'] = get_previous_close(day_k['code'], prev_trade_date)
                    else:
                        day_k['last_day_close'] = None

                batch_data.extend(day_k_list)
                total_processed += 1
                network_error_count = 0

                if len(batch_data) >= BATCH_SIZE:
                    inserted = batch_upsert_k_data(batch_data, 'day')
                    total_inserted += inserted
                    batch_data = []
                    logger.info(f"日K同步进度：{total_processed}/{len(stock_codes)}只股票")

                time.sleep(API_INTERVAL_DAY)

            except Exception as e:
                logger.error(f"处理{raw_code}日K失败：{str(e)}，重试1次")
                time.sleep(1)
                try:
                    day_k_list = fetch_stock_day_k_today(raw_code, today)
                    if day_k_list:
                        for day_k in day_k_list:
                            if prev_trade_date:
                                day_k['last_day_close'] = get_previous_close(day_k['code'], prev_trade_date)
                        batch_data.extend(day_k_list)
                        total_processed += 1
                    network_error_count = 0
                except Exception as e2:
                    logger.error(f"重试{raw_code}日K失败：{str(e2)}，放弃")
                    if '网络' in str(e2) or 'socket' in str(e2).lower():
                        network_error_count += 1
                continue

        if batch_data:
            inserted = batch_upsert_k_data(batch_data, 'day')
            total_inserted += inserted

        update_synced_date(today)

    except Exception as e:
        logger.error(f"日K同步任务异常终止：{str(e)}", exc_info=True)
    finally:
        baostock_safe_logout()

    elapsed = int(time.time() - start_time)
    total_records = get_total_records('day')
    logger.info(f"""
    ===== 日K同步任务完成 =====
    交易日：{today if 'today' in locals() else '未知'}
    总处理股票：{total_processed if 'total_processed' in locals() else 0}只
    成功插入/更新：{total_inserted if 'total_inserted' in locals() else 0}条日K数据
    数据库总记录：{total_records}条
    耗时：{elapsed // 3600}小时{(elapsed % 3600) // 60}分钟{elapsed % 60}秒
    """)


# 周K同步逻辑（优化only_full_group_by问题）
def sync_week_k():
    try:
        logger.info("===== 开始周K同步任务 =====\n")
        start_time = time.time()

        # 获取周参数
        week_params = get_week_params()
        current_week_start, current_week_end = week_params['current']
        last_week_end = week_params['last_week_end']
        week_k_date = week_params['week_end']
        is_week_complete = week_params['is_complete']

        logger.info(f"本周交易日范围：{current_week_start}至{current_week_end}，周K日期：{week_k_date}")

        # 检查是否已同步过该周期
        if os.path.exists(PROGRESS_FILE_WEEK):
            with open(PROGRESS_FILE_WEEK, 'r', encoding='utf-8') as f:
                if f.read().strip() == f"SYNCED:{current_week_start}_{current_week_end}":
                    logger.info(f"周K周期{current_week_start}-{current_week_end}已同步，跳过")
                    return

        # 执行SQL：一次性聚合所有股票的周K数据（解决only_full_group_by问题）
        conn = pymysql.connect(** DB_CONFIG)
        cursor = conn.cursor()

        # 核心SQL：子查询+聚合函数解决only_full_group_by问题
        week_sql = f"""
        INSERT INTO stock_week_k_history (
            id, code, date, open, high, low, close, 
            last_week_close, volume, amount, adjustflag, turn, pctChg
        )
        -- 外部查询：按code分组，用MAX确保符合only_full_group_by（子查询已保证每个code一行）
        SELECT 
            MAX(sub.id),
            sub.code,
            MAX(sub.week_k_date),
            MAX(sub.week_open),
            MAX(sub.week_high),
            MAX(sub.week_low),
            MAX(sub.week_close),
            MAX(sub.last_week_close),
            MAX(sub.week_volume),
            MAX(sub.week_amount),
            MAX(sub.adjustflag),
            MAX(sub.week_turn),
            MAX(sub.week_pctChg)
        FROM (
            -- 子查询：用窗口函数计算周指标，每个code生成一行数据
            SELECT 
                CONCAT(d.code, '_', REPLACE('{week_k_date}', '-', '')) AS id,
                d.code,
                '{week_k_date}' AS week_k_date,
                FIRST_VALUE(d.open) OVER (PARTITION BY d.code ORDER BY d.date) AS week_open,
                MAX(d.high) OVER (PARTITION BY d.code) AS week_high,
                MIN(d.low) OVER (PARTITION BY d.code) AS week_low,
                LAST_VALUE(d.close) OVER (
                    PARTITION BY d.code 
                    ORDER BY d.date 
                    ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
                ) AS week_close,
                (SELECT close FROM stock_day_k_history 
                 WHERE code = d.code AND date = '{last_week_end}') AS last_week_close,
                SUM(d.volume) OVER (PARTITION BY d.code) AS week_volume,
                SUM(d.amount) OVER (PARTITION BY d.code) AS week_amount,
                FIRST_VALUE(d.adjustflag) OVER (PARTITION BY d.code ORDER BY d.date) AS adjustflag,
                SUM(d.turn) OVER (PARTITION BY d.code) AS week_turn,
                (LAST_VALUE(d.close) OVER (
                    PARTITION BY d.code 
                    ORDER BY d.date 
                    ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
                 ) - (SELECT close FROM stock_day_k_history 
                      WHERE code = d.code AND date = '{last_week_end}')) 
                 / NULLIF((SELECT close FROM stock_day_k_history 
                           WHERE code = d.code AND date = '{last_week_end}'), 0) * 100 AS week_pctChg
            FROM stock_day_k_history d
            WHERE d.date BETWEEN '{current_week_start}' AND '{current_week_end}'
        ) AS sub
        GROUP BY sub.code;  -- 子查询已按code分区，外部按code分组确保唯一
        """

        try:
            # 先删除该周期已存在的周K记录（避免重复）
            cursor.execute(f"""
                DELETE FROM stock_week_k_history 
                WHERE date = '{week_k_date}'
            """)
            # 执行批量聚合插入（一条语句处理所有股票）
            affected_rows = cursor.execute(week_sql)
            conn.commit()
            logger.info(f"周K同步完成：{affected_rows}条记录（覆盖所有股票）")

            # 记录周期同步状态（非股票级）
            with open(PROGRESS_FILE_WEEK, 'w') as f:
                f.write(f"SYNCED:{current_week_start}_{current_week_end}")

        except Exception as e:
            conn.rollback()
            logger.error(f"周K SQL执行失败：{str(e)}", exc_info=True)
            return
        finally:
            conn.close()

    except Exception as e:
        logger.error(f"周K同步任务异常终止：{str(e)}", exc_info=True)

    elapsed = int(time.time() - start_time)
    total_records = get_total_records('week')
    logger.info(f"""
    ===== 周K线同步任务完成 =====
    周范围：{current_week_start}至{current_week_end}
    成功插入/更新：{affected_rows if 'affected_rows' in locals() else 0}条周K数据
    数据库总记录：{total_records}条
    耗时：{elapsed // 3600}小时{(elapsed % 3600) // 60}分钟{elapsed % 60}秒
    """)


# 月K同步逻辑（优化only_full_group_by问题）
def sync_month_k():
    try:
        logger.info("===== 开始月K同步任务 =====\n")
        start_time = time.time()

        # 获取月参数
        month_params = get_month_params()
        current_month_start, current_month_end = month_params['current']
        last_month_end = month_params['last_month_end']
        month_k_date = month_params['month_end']
        is_month_complete = month_params['is_complete']

        logger.info(f"本月交易日范围：{current_month_start}至{current_month_end}，月K日期：{month_k_date}")

        # 检查是否已同步过该周期
        if os.path.exists(PROGRESS_FILE_MONTH):
            with open(PROGRESS_FILE_MONTH, 'r', encoding='utf-8') as f:
                if f.read().strip() == f"SYNCED:{current_month_start}_{current_month_end}":
                    logger.info(f"月K周期{current_month_start}-{current_month_end}已同步，跳过")
                    return

        # 执行SQL：一次性聚合所有股票的月K数据（解决only_full_group_by问题）
        conn = pymysql.connect(**DB_CONFIG)
        cursor = conn.cursor()

        # 核心SQL：子查询+聚合函数解决only_full_group_by问题
        month_sql = f"""
        INSERT INTO stock_month_k_history (
            id, code, date, open, high, low, close, 
            last_month_close, volume, amount, adjustflag, turn, pctChg
        )
        -- 外部查询：按code分组，用MAX确保符合only_full_group_by
        SELECT 
            MAX(sub.id),
            sub.code,
            MAX(sub.month_k_date),
            MAX(sub.month_open),
            MAX(sub.month_high),
            MAX(sub.month_low),
            MAX(sub.month_close),
            MAX(sub.last_month_close),
            MAX(sub.month_volume),
            MAX(sub.month_amount),
            MAX(sub.adjustflag),
            MAX(sub.month_turn),
            MAX(sub.month_pctChg)
        FROM (
            -- 子查询：用窗口函数计算月指标，每个code生成一行数据
            SELECT 
                CONCAT(d.code, '_', REPLACE('{month_k_date}', '-', '')) AS id,
                d.code,
                '{month_k_date}' AS month_k_date,
                FIRST_VALUE(d.open) OVER (PARTITION BY d.code ORDER BY d.date) AS month_open,
                MAX(d.high) OVER (PARTITION BY d.code) AS month_high,
                MIN(d.low) OVER (PARTITION BY d.code) AS month_low,
                LAST_VALUE(d.close) OVER (
                    PARTITION BY d.code 
                    ORDER BY d.date 
                    ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
                ) AS month_close,
                (SELECT close FROM stock_day_k_history 
                 WHERE code = d.code AND date = '{last_month_end}') AS last_month_close,
                SUM(d.volume) OVER (PARTITION BY d.code) AS month_volume,
                SUM(d.amount) OVER (PARTITION BY d.code) AS month_amount,
                FIRST_VALUE(d.adjustflag) OVER (PARTITION BY d.code ORDER BY d.date) AS adjustflag,
                SUM(d.turn) OVER (PARTITION BY d.code) AS month_turn,
                (LAST_VALUE(d.close) OVER (
                    PARTITION BY d.code 
                    ORDER BY d.date 
                    ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
                 ) - (SELECT close FROM stock_day_k_history 
                      WHERE code = d.code AND date = '{last_month_end}')) 
                 / NULLIF((SELECT close FROM stock_day_k_history 
                           WHERE code = d.code AND date = '{last_month_end}'), 0) * 100 AS month_pctChg
            FROM stock_day_k_history d
            WHERE d.date BETWEEN '{current_month_start}' AND '{current_month_end}'
        ) AS sub
        GROUP BY sub.code;  -- 子查询已按code分区，外部按code分组确保唯一
        """

        try:
            # 先删除该周期已存在的月K记录
            cursor.execute(f"""
                DELETE FROM stock_month_k_history 
                WHERE date = '{month_k_date}'
            """)
            # 执行批量聚合插入（一条语句处理所有股票）
            affected_rows = cursor.execute(month_sql)
            conn.commit()
            logger.info(f"月K同步完成：{affected_rows}条记录（覆盖所有股票）")

            # 记录周期同步状态（非股票级）
            with open(PROGRESS_FILE_MONTH, 'w') as f:
                f.write(f"SYNCED:{current_month_start}_{current_month_end}")

        except Exception as e:
            conn.rollback()
            logger.error(f"月K SQL执行失败：{str(e)}", exc_info=True)
            return
        finally:
            conn.close()

    except Exception as e:
        logger.error(f"月K同步任务异常终止：{str(e)}", exc_info=True)

    elapsed = int(time.time() - start_time)
    total_records = get_total_records('month')
    logger.info(f"""
    ===== 月K线同步任务完成 =====
    月范围：{current_month_start}至{current_month_end}
    成功插入/更新：{affected_rows if 'affected_rows' in locals() else 0}条月K数据
    数据库总记录：{total_records}条
    耗时：{elapsed // 3600}小时{(elapsed % 3600) // 60}分钟{elapsed % 60}秒
    """)


# 数据库批量操作（仅日K需要）
def batch_upsert_k_data(data_list, k_type):
    if not data_list or k_type != 'day':
        return 0

    table = 'stock_day_k_history'
    insert_fields = [
        'id', 'code', 'date', 'open', 'high', 'low', 'close',
        'last_day_close', 'volume', 'amount', 'adjustflag', 'turn', 'pctChg'
    ]

    try:
        conn = pymysql.connect(**DB_CONFIG)
        cursor = conn.cursor()
        # 批量删除旧数据
        delete_pairs = [(d['code'], d['date']) for d in data_list]
        cursor.executemany(
            f"DELETE FROM {table} WHERE code = %s AND date = %s",
            delete_pairs
        )

        # 批量插入新数据
        sql = f"INSERT INTO {table} ({','.join(insert_fields)}) VALUES ({','.join(['%s'] * len(insert_fields))})"
        values = []
        for d in data_list:
            value_tuple = (
                d['id'], d['code'], d['date'],
                round(d['open'], 2) if d['open'] else None,
                round(d['high'], 2) if d['high'] else None,
                round(d['low'], 2) if d['low'] else None,
                round(d['close'], 2) if d['close'] else None,
                round(d['last_day_close'], 2) if d['last_day_close'] else None,
                round(d['volume'], 2) if d['volume'] else None,
                round(d['amount'], 2) if d['amount'] else None,
                d['adjustflag'],
                round(d['turn'], 2) if d['turn'] else None,
                round(d['pctChg'], 2) if d['pctChg'] else None
            )
            values.append(value_tuple)

        affected_rows = cursor.executemany(sql, values)
        conn.commit()
        logger.info(f"日K批量操作完成：删除{len(delete_pairs)}条旧数据，插入{affected_rows}条新数据")
        return affected_rows
    except Exception as e:
        conn.rollback()
        logger.error(f"日K批量操作失败：{str(e)}", exc_info=True)
        return 0
    finally:
        if 'conn' in locals():
            conn.close()


# 辅助工具函数
def get_total_records(k_type):
    table = {
        'day': 'stock_day_k_history',
        'week': 'stock_week_k_history',
        'month': 'stock_month_k_history'
    }[k_type]
    try:
        conn = pymysql.connect(** DB_CONFIG)
        cursor = conn.cursor()
        cursor.execute(f"SELECT COUNT(*) FROM {table}")
        return cursor.fetchone()[0]
    except Exception as e:
        logger.warning(f"查询{table}总记录数失败：{str(e)}")
        return -1
    finally:
        if 'conn' in locals():
            conn.close()


# 主调度逻辑
def sync_all_k():
    """整合同步：日K→周K→月K"""
    try:
        logger.info("=== 开始【日K→周K→月K】合并同步任务 ===")
        # sync_day_k()
        logger.info("=== 日K同步完成，开始周K同步 ===")
        # sync_week_k()
        logger.info("=== 周K同步完成，开始月K同步 ===")
        sync_month_k()
        logger.info("=== 【日K→周K→月K】合并同步任务全部完成 ===")
    except Exception as e:
        logger.error(f"合并同步任务异常终止：{str(e)}", exc_info=True)


def main():
    try:
        init_database()
        if is_cache_expired(STOCK_CODE_CACHE):
            update_stock_code_cache()
        sync_all_k()  # 首次运行同步一次

        # 定时任务
        scheduler = BackgroundScheduler(timezone='Asia/Shanghai')
        scheduler.add_job(
            sync_all_k,
            'cron',
            hour=17,
            minute=55,
            second=0,
            id='stock_k_sync_job',
            replace_existing=True
        )
        logger.info("定时任务已启动，每日17:55执行【日K→周K→月K】同步")
        scheduler.start()

        try:
            while True:
                time.sleep(60)
        except (KeyboardInterrupt, SystemExit):
            scheduler.shutdown()
            baostock_safe_logout()
            logger.info("程序手动终止，定时任务已停止")
    except Exception as e:
        logger.error(f"主程序异常终止：{str(e)}", exc_info=True)
        baostock_safe_logout()


if __name__ == "__main__":
    main()
